package com.pai4j.common.service.messagequeue.task;

import com.fasterxml.jackson.core.type.TypeReference;
import com.pai4j.common.enums.MessageQueueTaskStatusEnum;
import com.pai4j.common.enums.messagequeue.MessageQueueEnum;
import com.pai4j.common.enums.messagequeue.MessageQueueMethodEnum;
import com.pai4j.common.service.messagequeue.producer.MessageQueueProducer;
import com.pai4j.common.util.JsonUtil;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import com.pai4j.common.util.TransactionSyncManagerUtil;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;


public abstract class AbstractMQTaskService<E> {


    @Autowired
    private MessageQueueProducer<String, Long> messageQueueProducer;

    /**
     * 分批查询失败任务条数
     */
    protected static final int QUERY_MAX_SIZE = 1000;

    /**
     * 获取失败需要重新归队的任务列表，由各个服务自行实现
     *
     * @return
     */
    protected abstract List<MessageQueueTaskDTO> tasks();

    public int saveFailTask2DB(MessageQueueTaskDTO task) {
        if (task == null)  {
            return 0;
        }
        // TODO 基础参数校验
        return this.saveFailTask2DB(List.of(task));
    }

    /**
     * 批量保存MQ失败任务到数据
     *
     * @param tasks
     * @return
     */
    public abstract int saveFailTask2DB(List<MessageQueueTaskDTO> tasks);

    /**
     * 任务重新归队
     *
     */
    protected void backQueue(MessageQueueTaskDTO task) {

        String queueName = task.getTopicName();
        String message = task.getTaskJson();
        String methodName = task.getMethod();
        try {
            MessageQueueEnum queueEnum = MessageQueueEnum.getByQueue(queueName);
            if (MessageQueueMethodEnum.PRIORITY_SEND.getMethod().equals(methodName)) {
                // 优先级队列，走反射调用框架producer的对应的消息发送
                Method method = messageQueueProducer.getClass().getDeclaredMethod(
                        methodName, MessageQueueEnum.class, String.class, Object.class);
                method.invoke(messageQueueProducer, queueEnum, message, task.getPriority());
            } else if (MessageQueueMethodEnum.SEND.getMethod().equals(methodName)) {
                // 普通队列发送，走反射调用框架producer的对应的消息发送
                Method method = messageQueueProducer.getClass().getDeclaredMethod(
                        methodName, MessageQueueEnum.class, Object.class);
                method.invoke(messageQueueProducer, queueEnum, message);
            } else {
                // 不支持的模式，不需要归队
                // 邮件通知
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 更新任务状态
     *
     * @param taskId
     * @param status
     */
    protected abstract void updateTaskStatus(List<String> taskId, MessageQueueTaskStatusEnum status);


    protected List<MessageQueueTaskDTO> convert(List<E> tasks) {

        if (CollectionUtils.isEmpty(tasks)) {
            return Collections.emptyList();
        }
        return JsonUtil.fromJson(JsonUtil.toJsonString(tasks), new TypeReference<List<MessageQueueTaskDTO>>() {});
    }

    /**
     * 失败任务归队
     *
     */
    @Transactional
    public void doExecute() {
        // 1. 取失败任务列表
        List<MessageQueueTaskDTO> tasks = this.tasks();
        if (CollectionUtils.isEmpty(tasks)) {
            return;
        }
        // 2. 归队成功，修改任务状态（避免对重复消费）
        List<String> taskIds = tasks.stream().map(MessageQueueTaskDTO::getTaskId).toList();
        this.updateTaskStatus(taskIds, MessageQueueTaskStatusEnum.RUNNING);
        /**
         * 事务提交后，后置处理任务
         * ps：这afterCommit中的任务及时出现异常，主事务也不会回滚（已经主事务已提交）
         */
        TransactionSyncManagerUtil.afterCommit(() -> {
            List<String> backQueueFailTask = new ArrayList<>();
            // 3. 执行归队逻辑
            tasks.forEach(task -> {
                try {
                    this.backQueue(task);
                } catch (Exception e) {
                    // 处理部分归队成功、部分失败场景
                    // 主事务提交后，任务重新归队异常，我们需要回滚异常task状态
                    backQueueFailTask.add(task.getTaskId());
                }
            });

            if (CollectionUtils.isNotEmpty(backQueueFailTask)) {
                rollbackBackQueueFailTaskStatus(backQueueFailTask);
            }

        });
    }

    /**
     * 创建一个新的事务，来处理部分归队失败的任务状态回滚
     *
     */
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void rollbackBackQueueFailTaskStatus(List<String> backQueueFailTask) {

        updateTaskStatus(backQueueFailTask, MessageQueueTaskStatusEnum.ERROR);
    }

}
